package com.gl.saas.basis.server.kafka.producer.api;

import com.alibaba.fastjson.JSON;
import com.gl.saas.basis.server.kafka.producer.in.BasicIn;
import com.gl.saas.basis.server.kafka.producer.out.BasicOut;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author ZWW
 * @Date 2019年10月25日14:39:19
 */
@RestController
public class ProducerController {

    @Resource
    KafkaTemplate<String,String> kafkaTemplate;

    @PostMapping("producer")
    public BasicOut devProducer(@RequestBody BasicIn basicIn){

       if (basicIn.getKey()==null||basicIn.getKey().trim().equals("")){
           return  BasicOut.fail("key不能为空");
       }
        if (basicIn.getTopic()==null||basicIn.getTopic().trim().equals("")){
            return  BasicOut.fail("topic不能为空");
        }
        kafkaTemplate.send(basicIn.getTopic(),basicIn.getKey(), JSON.toJSONString(basicIn.getData()));
        return  BasicOut.ok();
    }


}
